import java.io.File;
import java.io.IOException;
import java.sql.SQLException;

import oracle.jdbc.aq.AQDequeueOptions;
import oracle.jdbc.aq.AQMessage;
import oracle.jdbc.aq.AQMessageProperties;
import oracle.jdbc.driver.OracleConnection;
import oracle.sql.ANYDATA;
import oracle.xdb.XMLType;

import org.apache.log4j.Logger;


public class AQDequeue extends OracleAQ {
	
	private Logger logger = Logger.getLogger( AQDequeue.class);
	
	public AQDequeue() {}
	
	public AQDequeue( String user, String password, String host, String port, String sid) {
		super( user, password, host, port, sid);
	
	}
	
	public AQDequeue( String user, String password, String url) {
		super( user, password, url);
	
	}
	
	public byte[] dequeue( String queueName, String queueType, String consumerName) throws SQLException {
		
		logger.info( "============= Dequeue start =============");
		
		this.connect();
		
		AQDequeueOptions deqopt = new AQDequeueOptions();
		deqopt.setRetrieveMessageId( true);
		
		if( consumerName != null) {
			deqopt.setConsumerName( consumerName);
		}
		
		// Connection Timeout (per seconds)
//		deqopt.setWait( 2);
		// Dequeue operation
		AQMessage msg = this.conn.dequeue( queueName, deqopt, queueType);
		
		// Print out the message that has been dequeued
		byte[] payload = msg.getPayload();
		byte[] msgId = msg.getMessageId();
		
		if( msgId != null) {
			String msgIdStr = AQByteUtil.byteBufferToHexString( msgId, 20);
			logger.info( "ID of message dequeued = " + msgIdStr);
		}
		
		AQMessageProperties msgProp = msg.getMessageProperties();
		logger.info( msgProp.toString());
		
		logger.info( "payload.length = " + payload.length + ", value = " + new String( payload));
		logger.info( "============= Dequeue done =============");
		
		return payload;

	}
	
	public byte[] dequeue( OracleConnection conn, String queueName, String queueType, String consumerName) throws SQLException {
		
		logger.info( "============= Dequeue start =============");
		
		AQDequeueOptions deqopt = new AQDequeueOptions();
		deqopt.setRetrieveMessageId( true);
		 
		if( consumerName != null) {
			deqopt.setConsumerName( consumerName);
		}
		
		// Dequeue operation
		AQMessage msg = conn.dequeue( queueName, deqopt, queueType);
		// Print out the message that has been dequeued
		
//		byte[] rawPayload = msg.getPayload();
//		InputStream is = new ByteArrayInputStream( rawPayload);
//		XMLType payload;
////		payload = XMLType.createXML( conn, is);
//		CLOB clob = new CLOB( conn, rawPayload);
//		payload = XMLType.createXML( conn, clob);
////		XMLType payload = msg.getXMLTypePayload();
		
//		ANYDATA any = msg.getANYDATAPayload();
////		System.out.println("Data : " + any.stringValue( conn));
////		this.cstmt = this.conn.prepareCall("{call ANYDATA.accessClob(?)}");
////		this.cstmt.setObject( 1, any);
//		
//		PreparedStatement pstmt = conn.prepareStatement( "SELECT SYS.ANYDATA.accessVarchar2(?) from dual");
//		pstmt.setObject( 1, any);
////		ResultSet rs = this.cstmt.executeQuery();
//		ResultSet rs = pstmt.executeQuery();
//		rs.next();
//		System.out.println( "Data : " + rs.getObject(1));
//		byte[] payload = any.getData();
//		System.out.println( "Data2 : " + new String(payload));
		
		ANYDATA any = msg.getANYDATAPayload();
//		TypeDescriptor td = any.getTypeDescriptor();
//		System.out.println( "getName = " + td.getName());
//		System.out.println( "getSchemaName = " + td.getSchemaName());
//		System.out.println( "getTypeName = " + td.getTypeName());
//		System.out.println( "getSQLName = " + td.getSQLName());
//		OpaqueDescriptor od = OpaqueDescriptor.createDescriptor(td.getSQLName(), conn);
//		System.out.println( any.stringValue());
//		XMLType xml = new XMLType(od, conn, any);
//		XMLType xml = XMLType.
//		System.out.println(xml.stringValue());
		
//		byte[] payload = msg.getPayload();
		byte[] payload = any.getData();
		System.out.println( "payload = " + new String( payload));
		
		byte[] msgId = msg.getMessageId();
		
		if( msgId != null) {
			String msgIdStr = AQByteUtil.byteBufferToHexString( msgId, 20);
			logger.info( "ID of message dequeued = " + msgIdStr);
		}
		
		AQMessageProperties msgProp = msg.getMessageProperties();
		logger.info( msgProp.toString());
		
//		logger.info( "payload.length = " + payload.length + ", value = " + new String( payload));
		logger.info( "============= Dequeue done =============");
		
		return payload;

	}
	
	public byte[] dequeueMulti( OracleConnection conn, String queueName, String queueType, String consumerName) throws SQLException {
		
		logger.info( "============= Dequeue start =============");
		
		AQDequeueOptions deqopt = new AQDequeueOptions();
		deqopt.setRetrieveMessageId( true);

//		deqopt.setVisibility(VisibilityOption.IMMEDIATE);
		
//		AQAgent ag = AQFactory.createAQAgent();
		
//		deqopt.setDequeueMode( DequeueMode.BROWSE);
		
		if( consumerName != null) {
			deqopt.setConsumerName( consumerName);
		}
		
		System.out.println( "ConsumerName = " + deqopt.getConsumerName());
		// Dequeue operation
		AQMessage msg = conn.dequeue( queueName, deqopt, queueType);
		// Print out the message that has been dequeued
//		byte[] rawPayload = msg.getPayload();
//		InputStream is = new ByteArrayInputStream( rawPayload);
//		XMLType payload;
////		payload = XMLType.createXML( conn, is);
//		CLOB clob = new CLOB( conn, rawPayload);
//		payload = XMLType.createXML( conn, clob);
////		XMLType payload = msg.getXMLTypePayload();
		
//		ANYDATA any = msg.getANYDATAPayload();
////		System.out.println("Data : " + any.stringValue( conn));
////		this.cstmt = this.conn.prepareCall("{call ANYDATA.accessClob(?)}");
////		this.cstmt.setObject( 1, any);
//		
//		PreparedStatement pstmt = conn.prepareStatement( "SELECT SYS.ANYDATA.accessVarchar2(?) from dual");
//		pstmt.setObject( 1, any);
////		ResultSet rs = this.cstmt.executeQuery();
//		ResultSet rs = pstmt.executeQuery();
//		rs.next();
//		System.out.println( "Data : " + rs.getObject(1));
//		byte[] payload = any.getData();
//		System.out.println( "Data2 : " + new String(payload));
		
		ANYDATA any = msg.getANYDATAPayload();
//		TypeDescriptor td = any.getTypeDescriptor();
//		System.out.println( "getName = " + td.getName());
//		System.out.println( "getSchemaName = " + td.getSchemaName());
//		System.out.println( "getTypeName = " + td.getTypeName());
//		System.out.println( "getSQLName = " + td.getSQLName());
//		OpaqueDescriptor od = OpaqueDescriptor.createDescriptor(td.getSQLName(), conn);
//		System.out.println( any.stringValue());
//		XMLType xml = new XMLType(od, conn, any);
//		XMLType xml = XMLType.
//		System.out.println(xml.stringValue());
		
//		byte[] payload = msg.getPayload();
		byte[] payload = any.getData();
		System.out.println( "payload = " + new String( payload));
		
		byte[] msgId = msg.getMessageId();
		
		if( msgId != null) {
			String msgIdStr = AQByteUtil.byteBufferToHexString( msgId, 20);
			logger.info( "ID of message dequeued = " + msgIdStr);
		}
		
		AQMessageProperties msgProp = msg.getMessageProperties();
		logger.info( msgProp.toString());
		
//		logger.info( "payload.length = " + payload.length + ", value = " + new String( payload));
		logger.info( "============= Dequeue done =============");
		
		return payload;

	}
	
	public byte[] dequeueXml( OracleConnection conn, String queueName, String queueType, String consumerName) throws SQLException {
		
		logger.info( "============= Dequeue start =============");
		
		AQDequeueOptions deqopt = new AQDequeueOptions();
		deqopt.setRetrieveMessageId( true);
		 
		if( consumerName != null) {
			deqopt.setConsumerName( consumerName);
		}
		
		// Dequeue operation
		AQMessage msg = conn.dequeue( queueName, deqopt, queueType);
		// Print out the message that has been dequeued
		
		XMLType xml = null;
		xml = msg.getXMLTypePayload();
		System.out.println( "getStringVal = " +xml.getStringVal());
//		System.out.println( "getBytesValue = " + new String(xml.getBytesValue()));
		System.out.println( "getDescriptor = " + xml.getDescriptor());
		System.out.println( "getSQLTypeName = " + xml.getSQLTypeName());
		
//		byte[] payload = msg.getPayload();
		byte[] payload = xml.getStringVal().getBytes();
		
//		System.out.println( new String( payload));
		byte[] msgId = msg.getMessageId();
		
		if( msgId != null) {
			String msgIdStr = AQByteUtil.byteBufferToHexString( msgId, 20);
			logger.info( "ID of message dequeued = " + msgIdStr);
		}
		
		AQMessageProperties msgProp = msg.getMessageProperties();
		logger.info( msgProp.toString());
		
//		logger.info( "payload.length = " + payload.length + ", value = " + new String( payload));
		logger.info( "============= Dequeue done =============");
		
		return payload;

	}
	
	public byte[] dequeueRaw( OracleConnection conn, String queueName, String queueType, String consumerName) throws SQLException {
		
		logger.info( "============= Dequeue start =============");
		
		AQDequeueOptions deqopt = new AQDequeueOptions();
		deqopt.setRetrieveMessageId( true);
//		deqopt.setDequeueMode( DequeueMode.BROWSE);
		
		if( consumerName != null) {
			deqopt.setConsumerName( consumerName);
		}
		
		// Dequeue operation
		AQMessage msg = conn.dequeue( queueName, deqopt, queueType);
		// Print out the message that has been dequeued
		
//		byte[] payload = msg.getPayload();
		byte[] payload = msg.getRAWPayload().getBytes();
		
		System.out.println( new String( payload));
		
		byte[] msgId = msg.getMessageId();
		
		if( msgId != null) {
			String msgIdStr = AQByteUtil.byteBufferToHexString( msgId, 20);
			logger.info( "ID of message dequeued = " + msgIdStr);
		}
		
		AQMessageProperties msgProp = msg.getMessageProperties();
		logger.info( msgProp.toString());
		
//		logger.info( "payload.length = " + payload.length + ", value = " + new String( payload));
		logger.info( "============= Dequeue done =============");
		
		return payload;

	}

	public String dequeueStr( String queueName, String queueType, String consumerName) throws SQLException {
		
		logger.info( "============= Dequeue start =============");
		
		this.connect();
		
		AQDequeueOptions deqopt = new AQDequeueOptions();
		deqopt.setRetrieveMessageId( true);
		
		if( consumerName != null) {
			deqopt.setConsumerName( consumerName);
		}
		
		// Dequeue operation
		AQMessage msg = this.conn.dequeue( queueName, deqopt, queueType);
		
		// Print out the message that has been dequeued
		byte[] payload = msg.getPayload();
		byte[] msgId = msg.getMessageId();
		
		if( msgId != null) {
			String msgIdStr = AQByteUtil.byteBufferToHexString( msgId, 20);
			logger.info( "ID of message dequeued = " + msgIdStr);
		}
		
		AQMessageProperties msgProp = msg.getMessageProperties();
		logger.info( msgProp.toString());
		
		String payloadStr = new String( payload);
		logger.info( "payload.length = " + payload.length + ", value = " + payloadStr);
		logger.info( "============= Dequeue done =============");
		
		return payloadStr;

	}
	
	public String dequeueStr( OracleConnection conn, String queueName, String queueType, String consumerName) throws SQLException {
		
		logger.info( "============= Dequeue start =============");
		
		AQDequeueOptions deqopt = new AQDequeueOptions();
		deqopt.setRetrieveMessageId( true);
		if( consumerName != null) {
			deqopt.setConsumerName( consumerName);
		}
		
		// Dequeue operation
		AQMessage msg = conn.dequeue( queueName, deqopt, queueType);
		
		// Print out the message that has been dequeued
		byte[] payload = msg.getPayload();
		byte[] msgId = msg.getMessageId();
		
		if( msgId != null) {
			String msgIdStr = AQByteUtil.byteBufferToHexString( msgId, 20);
			logger.info( "ID of message dequeued = " + msgIdStr);
		}
		
		AQMessageProperties msgProp = msg.getMessageProperties();
		logger.info( msgProp.toString());
		
		String payloadStr = new String( payload);
		logger.info( "payload.length = " + payload.length + ", value = " + payloadStr);
		logger.info( "============= Dequeue done =============");
		
		return payloadStr;

	}
	
	public static void main( String[] args) {
	
		AQDequeue test = new AQDequeue( "strmadmin", "streams", "192.168.0.98", "1521", "bizteco");
//		AQDequeue test = new AQDequeue( "strmadmin", "streams", "192.168.0.198", "1521", "bizteco");
//		AQDequeue test = new AQDequeue( "cdc", "btc7btc", "192.168.0.198", "1521", "bizteco");
		
		OracleConnection conn = test.connect();
		
		try {
//			org.apache.commons.io.FileUtils.writeByteArrayToFile( new File("D:\\test.txt"), test.dequeue( conn, "OGG_QUEUE", "SYS.ANYDATA", null));
//			org.apache.commons.io.FileUtils.writeByteArrayToFile( new File("D:\\test3.txt"), test.dequeueRaw( conn, "OGG_RAW_QUEUE", "RAW", null));
//			org.apache.commons.io.FileUtils.writeByteArrayToFile( new File("D:\\test2.txt"), test.dequeueXml( conn, "OGG_XML_QUEUE", "SYS.XMLTYPE", null));
//			org.apache.commons.io.FileUtils.writeByteArrayToFile( new File("D:\\test.txt"), test.dequeueMulti( conn, "Q_CAPTURE", "SYS.ANYDATA", "strmadmin"));
			org.apache.commons.io.FileUtils.writeByteArrayToFile( new File("D:\\test.txt"), test.dequeueMulti( conn, "STRMADMIN.Q_CAPTURE", "SYS.ANYDATA", "streams_queue_agent"));
//			org.apache.commons.io.FileUtils.writeByteArrayToFile( new File("D:\\test.txt"), test.dequeueMulti( conn, "STRMADMIN.Q_CAPTURE", "SYS.ANYDATA", "strmadmin_capture"));
//			org.apache.commons.io.FileUtils.writeByteArrayToFile( new File("D:\\test.txt"), test.dequeueMulti( conn, "STRMADMIN.Q_CAPTURE", "SYS.ANYDATA", "STREAMS_QUEUE_AGENT"));
//			org.apache.commons.io.FileUtils.writeByteArrayToFile( new File("D:\\test.txt"), test.dequeueMulti( conn, "Q_APPLY", "SYS.ANYDATA", null));

		} catch (SQLException e) {
			e.printStackTrace();

		} catch (IOException e) {
			e.printStackTrace();

		}
		
		test.disconnect();
	}
	
}
